package com.bungarus.group.cachestore;

import com.bungarus.group.exception.BungarusGroupRuntimeException;
import io.lettuce.core.*;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.async.RedisAsyncCommands;
import io.lettuce.core.api.sync.RedisCommands;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.env.Environment;
import org.springframework.data.domain.Pageable;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;

@Component
public class CacheStore {
    private static final int MAX_TRY_COUNT = 3;
    private static final long WAITING_TIMEOUT = 3;
    private static final String REDIS_URI = "bungarus.lettuce.redis.url";
    private static final String REDIS_DB = "bungarus.lettuce.redis.db";
    private static final String REDIS_COMMAND_TIMEOUT = "bungarus.lettuce.redis.timeout";

    private static final String REDIS_SENTINEL_URI = "bungarus.lettuce.redis.sentinel.url";
    private static final String REDIS_SENTINEL_DB = "bungarus.lettuce.redis.sentinel.db";
    private static final String REDIS_SENTINEL_TIMEOUT = "bungarus.lettuce.redis.sentinel.timeout";
    private static final String REDIS_SENTINEL_MASTER_ID = "bungarus.lettuce.redis.sentinel.masterid";

    private Environment environment;
    private RedisClient redisClient;
    private StatefulRedisConnection<String, String> connection = null;

    private long redisCommandTimeout;

    @Autowired
    public CacheStore(Environment environment) {
        this.environment = environment;
    }

    public static String ofUniqueId(String... subIds) {
        StringBuilder sb = new StringBuilder();
        for(String id: subIds) {
            sb.append(id);
            sb.append(StoreRow.SEPERATOR);
        }
        if(sb.length() > 1) {
            sb.deleteCharAt(sb.length() - StoreRow.SEPERATOR.length());
        }
        return sb.toString();
    }

    @PostConstruct
    public void init() {
        String standaloneUri = environment.getProperty(REDIS_URI, String.class, "");
        Integer standaloneDb = environment.getProperty(REDIS_DB, Integer.class, 0);
        Integer standaloneTimeout = environment.getProperty(REDIS_COMMAND_TIMEOUT, Integer.class, 10000);

        String sentinelUri = environment.getProperty(REDIS_SENTINEL_URI, String.class, "");
        String sentinelMasterId = environment.getProperty(REDIS_SENTINEL_MASTER_ID, String.class, "");
        Integer sentinelDb = environment.getProperty(REDIS_SENTINEL_DB, Integer.class, 0);
        Integer sentinelTimeout = environment.getProperty(REDIS_SENTINEL_TIMEOUT, Integer.class, 10000);

        RedisURI uri = null;
        if(standaloneUri.length() > 0) {
            uri = RedisURI.create(standaloneUri);
            uri.setDatabase(standaloneDb);
            redisCommandTimeout = standaloneTimeout;

        }
        else if(sentinelUri.length() > 0){
            uri = RedisURI.create(sentinelUri);
            uri.setDatabase(sentinelDb);
            uri.setSentinelMasterId(sentinelMasterId);
            redisCommandTimeout = sentinelTimeout;
        }
        else {
            throw new UnsupportedOperationException();
        }
        uri.setTimeout(Duration.of(redisCommandTimeout, ChronoUnit.MILLIS));

        redisClient = RedisClient.create(uri);
        redisClient.setOptions(ClientOptions.builder()
                        .autoReconnect(true)
                        .pingBeforeActivateConnection(true)
                        .socketOptions(SocketOptions.builder().tcpNoDelay(true).build())
                        .build());
        connection = redisClient.connect();
    }

    @PreDestroy
    private void destroy() {
        redisClient.shutdown();
    }

    private StatefulRedisConnection<String, String> getConnection() {
        return connection;
    }

    public <T> void save(CacheTransformer<T> transformer, T source) {
        List<StoreRow> rows = new ArrayList<>(1);
        rows.add(transformer.transform(source));
        saveList(rows, transformer.getCachedAction());
    }

    public <T> void saveList(CacheTransformer<T> transformer, T source) {
        saveList(transformer.transform2List(source), transformer.getCachedAction());
    }

    private void saveList(List<StoreRow> rows, int cacheAction) {
        StatefulRedisConnection<String, String> connection = getConnection();
        RedisAsyncCommands<String, String> asyncCommands = connection.async();
        asyncCommands.setAutoFlushCommands(false);
        saveWithRetry(0, asyncCommands, rows, cacheAction);
    }

    private void saveWithRetry(int totalCount, RedisAsyncCommands<String, String> asyncCommands,
                               List<StoreRow> rows, int cacheAction) {
        if(totalCount >= MAX_TRY_COUNT) {
            throw new BungarusGroupRuntimeException("Try failed => Exceed " + MAX_TRY_COUNT + " TIMES.");
        }
        List<RedisFuture<?>> futures = new ArrayList<>();
        asyncCommands.multi();
        rows.forEach(row -> execRedisCommandInBuffer(futures, asyncCommands, row, cacheAction));
        asyncCommands.exec();
        asyncCommands.flushCommands();
        boolean result = false;
        try {
            result = LettuceFutures.awaitAll(WAITING_TIMEOUT, TimeUnit.SECONDS, futures.toArray(new RedisFuture[futures.size()]));
        }
        finally {
            futures.clear();
            if (!result) {
                try {
                    Thread.sleep(300);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                saveWithRetry(totalCount + 1, asyncCommands, rows, cacheAction);
            }
        }
    }

    public <T> T findByUniqueId(String uniqueId, CacheTransformer<T> transformer) {
        Set<StoreColumn> columns = null;

        StatefulRedisConnection<String, String> connection = getConnection();
        RedisAsyncCommands<String, String> asyncCommands = connection.async();

        String cacheKey = ofUniqueId(transformer.getCacheTableName(), uniqueId);

        if(CacheTransformer.CACHE_ACTION_HSET == transformer.getCachedAction()) {
            RedisFuture<Map<String, String>> mediumFuture = asyncCommands.hgetall(cacheKey);
            connection.flushCommands();

            Map<String, String> medium = getFutureValue(mediumFuture);

            if(null == medium || medium.size() == 0) {
                return null;
            }

            columns = medium.entrySet().stream()
                        .map(entry -> {
                            StoreColumn column = new StoreColumn();
                            column.setColumnName(entry.getKey());
                            column.setColumnValue(entry.getValue());
                            return column;
                        })
                        .collect(Collectors.toSet());

        }
        else if(CacheTransformer.CACHE_ACTION_SORTED_SET == transformer.getCachedAction()) {
            RedisFuture<List<ScoredValue<String>>> scoredValueFuture = asyncCommands.zrangeWithScores(cacheKey, 0, -1);
            connection.flushCommands();

            List<ScoredValue<String>> scoredValues = getFutureValue(scoredValueFuture);
            if(null == scoredValues) {
                return null;
            }

            columns = scoredValues.stream()
                    .map(scoredValue -> {
                        StoreColumn column = new StoreColumn();
                        column.setColumnName(scoredValue.getValue());
                        column.setColumnValue(String.valueOf(scoredValue.getScore()));
                        return column;
                    })
                    .collect(Collectors.toSet());
        }
        StoreRow row = StoreRow.newBuilder()
                .table(transformer.getCacheTableName())
                .id(uniqueId)
                .values(columns)
                .build();

        return transformer.reverse(row);
    }

    public long itemCount(String uniqueId, CacheTransformer transformer) {
        StatefulRedisConnection<String, String> connection = getConnection();
        RedisAsyncCommands<String, String> asynccommands = connection.async();

        RedisFuture<Long> totalFuture = null;
        if(CacheTransformer.CACHE_ACTION_SORTED_SET == transformer.getCachedAction()) {
            totalFuture = asynccommands.zcard(ofUniqueId(transformer.getCacheTableName(), uniqueId));
        }
        else if(CacheTransformer.CACHE_ACTION_HSET == transformer.getCachedAction()) {
            totalFuture = asynccommands.hlen(ofUniqueId(transformer.getCacheTableName(), uniqueId));
        }
        connection.flushCommands();

        if(null != totalFuture) {
            Long result = getFutureValue(totalFuture);
            return result == null ? 0L : result;
        }
        return 0L;
    }

    public List<Long> items(String uniqueId, Pageable page, CacheTransformer transformer) {
        if(CacheTransformer.CACHE_ACTION_HSET == transformer.getCachedAction()) {
            throw new IllegalArgumentException("The action in CacheTransformer: " + transformer.getClass().getCanonicalName() + " is not supported!");
        }
        StatefulRedisConnection<String, String> connection = getConnection();
        RedisAsyncCommands<String, String> asyncCommands = connection.async();
        //low and hight are inclusive ranges, and low is zero based.
        long low = page.getOffset();
        long hight = low + page.getPageSize() - 1;

        RedisFuture<List<String>> valuesFuture = asyncCommands.zrange(
                                CacheStore.ofUniqueId(transformer.getCacheTableName(), uniqueId),
                                low, hight);
        connection.flushCommands();

        List<String> values = getFutureValue(valuesFuture);
        if(null != values) {
            return values.stream().map(Long::valueOf).collect(Collectors.toList());
        }
        return Collections.emptyList();
    }

    public boolean itemExists(String uniqueId, String item, CacheTransformer transformer) {
        StatefulRedisConnection<String, String> connection = getConnection();
        RedisAsyncCommands<String, String> asyncCommands = connection.async();

        boolean result = false;
        if(CacheTransformer.CACHE_ACTION_SORTED_SET == transformer.getCachedAction()) {
            RedisFuture<Double> scoreFuture = asyncCommands.zscore(
                    CacheStore.ofUniqueId(transformer.getCacheTableName(), uniqueId),
                    item);
            connection.flushCommands();
            Double score = getFutureValue(scoreFuture);
            result = score != null;
        }
        else if(CacheTransformer.CACHE_ACTION_HSET == transformer.getCachedAction()){
            RedisFuture<Boolean> valueFuture = asyncCommands.hexists(CacheStore.ofUniqueId(transformer.getCacheTableName(), uniqueId), item);
            connection.flushCommands();
            Boolean temp = getFutureValue(valueFuture);
            result = temp != null ? temp : false;
        }
        return result;
    }

    public boolean delete(String uniqueId, CacheTransformer transformer) {
        StatefulRedisConnection<String, String> connection = getConnection();
        RedisAsyncCommands<String, String> asyncCommands = connection.async();
        String cacheKey = ofUniqueId(transformer.getCacheTableName(), uniqueId);

        long currentTimestamp = Calendar.getInstance().getTimeInMillis() - 10000L;
        RedisFuture<Boolean> expireFuture = asyncCommands.pexpireat(cacheKey, currentTimestamp);
        connection.flushCommands();
        Boolean temp = getFutureValue(expireFuture);
        return temp != null ? temp: false;
    }

    public void deleteItem(String ofUniqueId, StoreColumn column, CacheTransformer cacheTransformer) {
        String cacheKey = ofUniqueId(cacheTransformer.getCacheTableName(), ofUniqueId);
        StatefulRedisConnection<String, String> connection = getConnection();
        RedisAsyncCommands<String, String> asyncCommands = connection.async();

        if(CacheTransformer.CACHE_ACTION_SORTED_SET == cacheTransformer.getCachedAction()) {
            asyncCommands.zrem(cacheKey, column.getColumnName());
        }
        else if(CacheTransformer.CACHE_ACTION_HSET == cacheTransformer.getCachedAction()){
            asyncCommands.hdel(cacheKey, column.getColumnName());
        }

        connection.flushCommands();
    }

    public <T> void deleteItems(CacheTransformer<T> transformer, T source) {
        List<StoreRow> rows = transformer.transform2List(source);
        if(null == rows || rows.size() == 0) {
            return;
        }
        StatefulRedisConnection<String, String> connection = getConnection();
        RedisAsyncCommands<String, String> asyncCommands = connection.async();
        asyncCommands.setAutoFlushCommands(false);
        asyncCommands.multi();
        rows.forEach(row -> {
            String cacheKey = row.getCacheKey();
            List<String> members = row.getRowValues().stream()
                    .map(StoreColumn::getColumnName)
                    .collect(Collectors.toList());
            if(CacheTransformer.CACHE_ACTION_SORTED_SET == transformer.getCachedAction()) {
                asyncCommands.zrem(cacheKey, (String[]) members.toArray());
            }
            else if(CacheTransformer.CACHE_ACTION_HSET == transformer.getCachedAction()){
                asyncCommands.hdel(cacheKey, (String[]) members.toArray());
            }
        });
        asyncCommands.exec();
        asyncCommands.flushCommands();
    }

    public <T extends CacheKeyValue> void increaseValueOfItemBy(final CacheTransformer transformer, final T source, int step) {
        StatefulRedisConnection<String, String> connection = getConnection();
        RedisCommands<String, String> command = connection.sync();
        if(CacheTransformer.CACHE_ACTION_HSET == transformer.getCachedAction()) {
            command.hincrby(source.getKey(), source.getField(), step);
        }
        else if(CacheTransformer.CACHE_ACTION_SORTED_SET == transformer.getCachedAction()) {
            command.zincrby(source.getKey(), Double.parseDouble(source.getKey()), source.getValue());
        }
    }

    public Object execSyncCommandsWithCallback(int action, CacheStoreCallback callback) {
        StatefulRedisConnection<String, String> connection = getConnection();
        RedisCommands<String, String> command = connection.sync();
        connection.setAutoFlushCommands(true);
        if(CacheTransformer.CACHE_ACTION_HSET == action) {
            return callback.callHset(command);
        }
        else if(CacheTransformer.CACHE_ACTION_SORTED_SET == action) {
            return callback.callZset(command);
        }
        else {
            throw new UnsupportedOperationException("Action is not supported.");
        }
    }

    private void execRedisCommandInBuffer(List<RedisFuture<?>> futures, RedisAsyncCommands<String, String> asyncCommands,
                                          StoreRow row, int cacheAction) {
        String cacheKey = row.getCacheKey();
        switch (cacheAction) {
            case CacheTransformer.CACHE_ACTION_HSET:
                Map<String, String> valuesMap = row.getRowValues().stream()
                        .collect(Collectors.toMap(StoreColumn::getColumnName, StoreColumn::getColumnValue));
                futures.add(asyncCommands.hmset(cacheKey, valuesMap));
                break;
            case CacheTransformer.CACHE_ACTION_SORTED_SET:
                List<ScoredValue<String>> scoredValues = row.getRowValues().stream()
                        .map(storeColumn -> {
                            String value = storeColumn.getColumnValue();
                            try {
                                long lvalue = Long.parseLong(value);
                                return ScoredValue.just(lvalue, storeColumn.getColumnName());
                            }
                            catch (Exception e) {
                                return ScoredValue.just(storeColumn.getColumnValue().hashCode(), storeColumn.getColumnName());
                            }
                        })
                        .collect(Collectors.toList());
                futures.add(asyncCommands.zadd(cacheKey, scoredValues.toArray()));
                break;
            default:
                break;
        }
    }

    private <V> V getFutureValue(RedisFuture<V> future) {
        try {
            return future.get(redisCommandTimeout, TimeUnit.MILLISECONDS);
        } catch (InterruptedException | ExecutionException | TimeoutException e) {
            e.printStackTrace();
        }
        return null;
    }
}
